/*
 * Copyright (c) 2011 Salzburg Research.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package at.newmedialab.skwrl.reasoner;

import at.newmedialab.skwrl.api.SKWRLBackend;
import at.newmedialab.skwrl.api.TripleUpdateListener;
import at.newmedialab.skwrl.model.Justification;
import at.newmedialab.skwrl.model.Pattern;
import at.newmedialab.skwrl.model.Program;
import at.newmedialab.skwrl.model.Rule;
import at.newmedialab.skwrl.parser.ParseException;
import kiwi.core.api.config.ConfigurationService;
import kiwi.core.api.persistence.PersistenceService;
import kiwi.core.api.query.QueryResult;
import kiwi.core.api.query.QueryService;
import kiwi.core.api.task.TaskManagerService;
import kiwi.core.api.transaction.Transaction;
import kiwi.core.api.transaction.TransactionData;
import kiwi.core.api.transaction.TransactionService;
import kiwi.core.api.triplestore.KnowledgeSpaceService;
import kiwi.core.api.triplestore.TripleStore;
import kiwi.core.model.query.LiteralField;
import kiwi.core.model.query.Pattern;
import kiwi.core.model.query.ResourceField;
import kiwi.core.model.query.VariableField;
import kiwi.core.model.rdf.KiWiNode;
import kiwi.core.model.rdf.KiWiResource;
import kiwi.core.model.rdf.KiWiTriple;
import kiwi.core.model.rdf.KiWiUriResource;
import kiwi.core.qualifiers.event.transaction.AfterCommit;
import kiwi.reasoner.api.program.ReasonerProgramService;
import kiwi.reasoner.model.Justification;
import kiwi.reasoner.model.Program;
import kiwi.reasoner.model.Rule;
import kiwi.reasoner.parser.ParseException;
import org.apache.commons.configuration.Configuration;
import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.configuration.PropertiesConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.enterprise.context.ApplicationScoped;
import javax.enterprise.event.Observes;
import javax.inject.Inject;
import javax.inject.Named;
import javax.persistence.EntityManager;
import javax.persistence.Query;
import javax.persistence.TypedQuery;
import java.util.*;
import java.util.concurrent.*;

/**
 * Add file description here!
 * <p/>
 * User: sschaffe
 */
public class ReasoningEngine<Triple,Node> implements TripleUpdateListener<Triple> {

	private static final Logger log = LoggerFactory.getLogger(ReasoningEngine.class);

	private Map<Pattern<Node>,Rule<Node>> patternRuleMap;

	private ExecutorService reasoners;

	private List<Program<Node>> programs;

	private static long taskCounter = 0;

    private Configuration config;

    /**
     * The backend implementation on top of which this reasoner is running.
     */
    private SKWRLBackend<Triple,Node> backend;

    public ReasoningEngine(SKWRLBackend<Triple,Node> backend) {
        this.backend   = backend;
        backend.addTripleUpdateListener(this);

        patternRuleMap = new HashMap<Pattern<Node>, Rule<Node>>();

        try {
            config = new PropertiesConfiguration("skwrl-config.properties");
        } catch (ConfigurationException e) {
            log.warn("could not load configuration file skwrl-config.properties from current directory, home directory, or classpath");
        }

        programs = new LinkedList<Program<Node>>();

		// create a threadpool and initialize it with a blocking queue that blocks when it is full to
		// not get into a resource leak
		reasoners = new ThreadPoolExecutor(
				1,config.getInt("reasoning.threads",1),5, TimeUnit.MINUTES,
				new ArrayBlockingQueue<Runnable>(config.getInt("reasoning.queuesize",10))
				);

    }


    /**
     * Add a reasoning program to the reasoning engine. Executions of the reasoning engine
     * will take this program into account. If the program should be applied to existing
     * triples, full reasoning needs to be started manually.
     *
     * @param program
     */
    public void addProgram(Program<Node> program) {
        if(!programs.contains(program)) {
            programs.add(program);

            for(Rule<Node> rule : program.getRules()) {
                for(Pattern<Node> pattern : rule.getBody()) {
                    patternRuleMap.put(pattern,rule);
                }
            }
        }
    }

    /**
     * Remove a reasoning program from the reasoning engine. Triples inferred by this program
     * will currently not be removed automatically.
     * @param program
     */
    public void removeProgram(Program<Node> program) {
        if(programs.contains(program)) {
            programs.remove(program);

            for(Rule<Node> rule : program.getRules()) {
                for(Pattern<Node> pattern : rule.getBody()) {
                    patternRuleMap.remove(pattern);
                }
            }
        }
    }



	/**
     * Process a triple update to the RDF backend with the added and removed triples
     * passed as argument. Incrementally apply the updates that are contained in the update
	 * to the reasoning programs that are in the system.
	 *
     * @param addedTriples   the triples that have been added during the update
     * @param removedTriples the triples that have been removed during the update
	 */
    @Override
    public void processTripleUpdate(final Set<Triple> addedTriples, final Set<Triple> removedTriples) {
		if( (addedTriples.size() > 0 || removedTriples.size() > 0) && patternRuleMap.size() > 0) {

			final String taskName = "Reasoner Task "+ ++taskCounter + " (incremental reasoning, "+addedTriples.size()+"/"+removedTriples.size()+ " added/removed triples)";

			Runnable indexer = new Runnable() {
				@Override
				public void run() {
                    log.info("Asynchronous {} started", taskName);
                    Thread.currentThread().setName("Asynchronous " + taskName);
					incrementalReasoning(addedTriples,removedTriples);
                    log.info("Asynchronous {} finished", taskName);
				}
			};
			try {
				reasoners.execute(indexer);
			} catch(RejectedExecutionException ex) {
                log.info("Synchronous {} started", taskName);
                incrementalReasoning(addedTriples, removedTriples);
                log.info("Synchronous {} finished", taskName);
			}


		}
	}


	private void incrementalReasoning(Set<Triple> addedTriples, Set<Triple> removedTriples) {
		// evaluate the rules for all added triples
		if(addedTriples.size() > 0) {
			long start2 = System.currentTimeMillis();
			processRules(addedTriples);
			log.info("REASONER: reasoning for {} new triples took {} ms overall",addedTriples.size(),System.currentTimeMillis()-start2);
		}

		if(removedTriples.size() > 0) {
			// first clean up justifications that are no longer supported
			cleanupJustifications(removedTriples);

			// then remove all inferred triples that are no longer supported
			cleanupUnsupported();

			// and finally garbage collect those triples that are inferred and deleted
			garbageCollectTriples();
		}

	}

	/**
	 * Clean all inferred triples and re-run all reasoning rules.
	 */
	public void reRunPrograms() {
			final String taskName = "Reasoner Task "+ ++taskCounter + " (full reasoning)";

			Runnable indexer = new Runnable() {
				@Override
				public void run() {
					log.info("Asynchronous {} started", taskName);
                    Thread.currentThread().setName("Asynchronous " + taskName);
					executeReasoner();
                    log.info("Asynchronous {} finished", taskName);
				}
			};
			try {
				reasoners.execute(indexer);
			} catch(RejectedExecutionException ex) {
                log.info("Synchronous {} started", taskName);
				executeReasoner();
                log.info("Synchronous {} finished", taskName);
			}
	}


	private void executeReasoner() {

        // clean up existing justifications, since we to a full reasoning again
		log.debug("removing old justifications");
        backend.clearJustifications();

        // clean up all inferred triples
        log.debug("cleaning up old inferred triples ...");
        backend.removeInferredTriples();


		Set<Justification<Triple>> justifications = new HashSet<Justification<Triple>>();

		// process the rules over the whole dataset
		log.debug("processing rules ...");
        for(Program<Node> p : programs) {
            for(Rule<Node> rule : p.getRules()) {
                processRule(rule,null,null,justifications);
            }
        }


		// persist the justifications that have been created in the rule processing
        backend.addJustifications(justifications);
	}


	/**
	 * This method iterates over all triples that are passed as argument and
	 * checks whether they are used as supporting triples justifications. All
	 * such justifications are removed. Triples that are no longer supported
	 * will later be cleaned up by {@link #cleanupUnsupported()}
	 * 
	 * @param removedTriples
	 */
	private void cleanupJustifications(Set<Triple> removedTriples) {
		taskManagerService.updateStatus("cleaning up justifications for "+removedTriples.size()+" removed triples");

		EntityManager em = persistenceService.getEntityManager();

		TypedQuery<Justification> q = em.createNamedQuery("reasoner.listJustificationsByBaseTriple", Justification.class);

		em.getTransaction().begin();

		for(KiWiTriple triple : removedTriples) {
			q.setParameter("triple",triple);

			List<Justification> justifications = q.getResultList();
			for(Justification j : justifications) {
				em.remove(j);
			}
			taskManagerService.progressTask();
		}

		em.flush();
		em.getTransaction().commit();
	}


	/**
	 * Cleanup inferred triples that are no longer supported by any justification.
	 */
	private void cleanupUnsupported() {
		taskManagerService.updateStatus("cleaning up unsupported triples");

		EntityManager em = persistenceService.getEntityManager();

		em.getTransaction().begin();
		TypedQuery<KiWiTriple> q = em.createNamedQuery("reasoner.listUnsupportedTriples", KiWiTriple.class);
		q.setParameter("inferred",knowledgeSpaceService.getInferredKnowledgeSpace());

		List<KiWiTriple> triples =  q.getResultList();

		if(triples.size() > 0) {
			log.debug("removing {} unsupported triples ...",triples.size());

			Transaction tx = transactionService.getTransaction();
			transactionService.begin(tx);
			try {
				for(KiWiTriple triple : triples) {
					tripleStore.removeTriple(triple);
				}
				transactionService.commit(tx);
			} catch(Exception ex) {
				log.error("error while removing unsupported triples",ex);

				transactionService.abort(tx);
				transactionService.rollback(tx);
			}
		}
	}


	private void garbageCollectTriples() {
		taskManagerService.updateStatus("garbage collection");

		EntityManager em = persistenceService.getEntityManager();

		em.getTransaction().begin();
		Query q = em.createNamedQuery("reasoner.garbageCollectTriples");
		int count = q.executeUpdate();
		em.getTransaction().commit();

		log.info("garbage collected {} triples", count);

	}

	/**
	 *
	 * @param addedTriples
	 */
	private void processRules(Set<Triple> addedTriples) {
		// TODO: report progress via TaskManagerService
		Transaction tx = transactionService.getTransaction();

		Set<Justification> justifications = new HashSet<Justification>();

		transactionService.begin(tx);

		try {
			taskManagerService.updateStatus("processing rules ...");
			// select the rules that have at least one matching pattern; the match method will
			// return a set of variable bindings that we will be used to prepopulate the bindings
			for(Pattern pattern : patternRuleMap.keySet()) {
				for(KiWiTriple triple : addedTriples) {
					QueryResult match = matches(pattern,triple);
					if(match != null) {
						log.debug("pattern {} matched with triple {}",pattern.toString(),triple.toN3());

						processRule(patternRuleMap.get(pattern), match, pattern, justifications);
					}
				}

			}

			transactionService.commit(tx);

			log.debug("adding {} justifications",justifications.size());

			taskManagerService.updateStatus("storing justifications ...");
			Set<Justification> baseJustifications = new HashSet<Justification>();
			Map<KiWiTriple,Collection<Justification>> justificationCache = new HashMap<KiWiTriple, Collection<Justification>>();

			for(Justification justification : justifications) {
				KiWiTriple triple = justification.getTriple();

				Justification newJustification = new Justification();
				newJustification.setSupportingRules(justification.getSupportingRules());
				newJustification.setTriple(triple);

				Set<Justification> tripleJustifications = Collections.singleton(newJustification);

				// resolve inferred triples by replacing them by their justifications
				for(KiWiTriple support : justification.getSupportingTriples()) {
					if(support.isInferred()) {
						Collection<Justification> supportJustifications = justificationCache.get(support);
						// cache justifications of triple in case they are needed again in this run
						if(supportJustifications == null) {
							supportJustifications = getJustifications(support, baseJustifications);
							justificationCache.put(support,supportJustifications);
						}

						if(supportJustifications.size() == 0) {
							log.error("error: inferred triple {} is not justified!",support);
						}

						// mix the two sets
						Set<Justification> oldTripleJustifications = tripleJustifications;
						tripleJustifications = new HashSet<Justification>();
						for(Justification j1 : oldTripleJustifications) {
							for(Justification j2 : supportJustifications) {
								Justification j3 = new Justification();
								j3.setTriple(triple);
								j3.getSupportingTriples().addAll(j1.getSupportingTriples());
								j3.getSupportingTriples().addAll(j2.getSupportingTriples());
								j3.getSupportingRules().addAll(j1.getSupportingRules());
								j3.getSupportingRules().addAll(j2.getSupportingRules());
								tripleJustifications.add(j3);
							}
						}
					} else {
						for(Justification j : tripleJustifications) {
							j.getSupportingTriples().add(support);
						}
					}
				}

				baseJustifications.addAll(tripleJustifications);
			}

			if(configurationService.getBooleanConfiguration("reasoning.remove_duplicate_justifications")) {
				// remove duplicate justifications
				justificationCache = new HashMap<KiWiTriple, Collection<Justification>>();
				for(Iterator<Justification> it = baseJustifications.iterator(); it.hasNext(); ) {
					Justification j = it.next();

					Collection<Justification> supportJustifications = justificationCache.get(j.getTriple());
					// cache justifications of triple in case they are needed again in this run
					if(supportJustifications == null) {
						supportJustifications = getJustifications(j.getTriple(), Collections.<Justification>emptySet());
						justificationCache.put(j.getTriple(),supportJustifications);
					}

					if(supportJustifications.contains(j)) {
						it.remove();
					}
				}
			}



			log.debug("{} justifications added after resolving inferred triples", baseJustifications.size());


			// persist the justifications that have been created in the rule processing
			EntityManager em = persistenceService.getEntityManager();
			em.getTransaction().begin();
			for(Justification j : baseJustifications) {
				em.persist(j);
			}
			em.flush();
			em.getTransaction().commit();

		} catch(Exception ex) {
			log.error("error while processing rules",ex);

			transactionService.abort(tx);
			transactionService.rollback(tx);
		}
	}

	/**
	 * Process the rule given as argument. The set of bindings passed as argument is used as a seed of
	 * bindings and will be further populated by pattern matches. The set of justifications is passed over
	 * from previous calls so that justifications can be persisted in a batch.
	 *
	 * @param rule
	 * @param match
	 * @param justifications
	 */
	private void processRule(Rule rule, QueryResult match, Pattern p, Set<Justification> justifications) {

		// get the variable bindings for the rule evaluation
		log.debug("REASONER: evaluating rule body for rule {} ...", rule);

		// create a collection consisting of the body minus the pattern that already matched
		Set<Pattern> body = new HashSet<Pattern>(rule.getBody());

		if(p != null) {
			body.remove(p);
		}

		List<QueryResult> bodyResult;

		// if there are further patterns, evaluate them; if the matched pattern was the only pattern, then
		// simply take the match as binding
		if(body.size() > 0) {
			bodyResult = queryService.query(body,match,null,null,0,0,true,true);
		} else if(match != null) {
			bodyResult = Collections.singletonList(match);
		} else {
			bodyResult = Collections.emptyList();
		}

		log.debug("REASONER: rule body evaluation yielded {} bindings", bodyResult.size());

		// construct triples out of the bindings and the rule heads
		for(QueryResult row : bodyResult) {
			Map<VariableField,KiWiNode> binding = row.getBindings();

			KiWiResource subject = null;
			KiWiUriResource property = null, context = null;
			KiWiNode object;

			if(rule.getHead().getSubject() != null && rule.getHead().getSubject().isVariableField()) {
				if(!binding.get(rule.getHead().getSubject()).isUriResource() && !binding.get(rule.getHead().getSubject()).isAnonymousResource()) {
					log.info("cannot use value {} as subject, because it is not a resource",binding.get(rule.getHead().getSubject()));
					continue;
				}
				subject = (KiWiResource)binding.get(rule.getHead().getSubject());
			} else if(rule.getHead().getSubject() != null && rule.getHead().getSubject().isResourceField()) {
				subject = ((ResourceField)rule.getHead().getSubject()).getResource();
			} else {
				throw new IllegalArgumentException("Subject of rule head may only be a variable or a resource; rule: "+rule);
			}

			if(rule.getHead().getProperty() != null && rule.getHead().getProperty().isVariableField()) {
				if(!binding.get(rule.getHead().getProperty()).isUriResource()) {
					log.info("cannot use value {} as property, because it is not a URI resource",binding.get(rule.getHead().getProperty()));
					continue;
				}
				property = (KiWiUriResource)binding.get(rule.getHead().getProperty());
			} else if(rule.getHead().getProperty() != null && rule.getHead().getProperty().isResourceField()) {
				property = (KiWiUriResource)((ResourceField)rule.getHead().getProperty()).getResource();
			} else {
				throw new IllegalArgumentException("Property of rule head may only be a variable or a resource; rule: "+rule);
			}

			if(rule.getHead().getObject() != null && rule.getHead().getObject().isVariableField()) {
				object = binding.get(rule.getHead().getObject());
			} else if(rule.getHead().getObject() != null && rule.getHead().getObject().isResourceField()) {
				object = ((ResourceField)rule.getHead().getObject()).getResource();
			} else if(rule.getHead().getObject() != null && rule.getHead().getObject().isLiteralField()) {
				object = ((LiteralField)rule.getHead().getObject()).getLiteral();
			} else {
				throw new IllegalArgumentException("Object of rule head may only be a variable, a literal, or a resource; rule: "+rule);
			}



			context = knowledgeSpaceService.getInferredKnowledgeSpace();

			KiWiTriple triple = tripleStore.createTriple(subject,property,object,context);
			triple.setInferred(true);

			Justification justification = new Justification();
			justification.setTriple(triple);
			justification.getSupportingRules().add(rule);
			justification.getSupportingTriples().addAll(row.getJustifications());
			justifications.add(justification);
		}

	}

	/**
	 * Return the justifications for the triple passed as argument.
	 * @param t
	 * @return
	 */
	private Collection<Justification> getJustifications(KiWiTriple t, Set<Justification> transactionJustifications) {
		Set<Justification> justifications = new HashSet<Justification>();

		EntityManager em = persistenceService.getEntityManager();

		TypedQuery<Justification> q = em.createNamedQuery("reasoner.listJustificationsBySupportedTriple", Justification.class);
		q.setParameter("triple",t);
		q.setHint("org.hibernate.cacheable", true);

		justifications.addAll(q.getResultList());

		for(Justification j : transactionJustifications) {
			if(j.getTriple() == t) {
				justifications.add(j);
			}
		}

		return justifications;
	}


	private QueryResult matches(Pattern pattern, KiWiTriple triple) {
		boolean result = true;

		QueryResult match = new QueryResult();

		if(pattern.getSubject().isResourceField()) {
			result = ((ResourceField)pattern.getSubject()).getResource().equals(triple.getSubject());
		} else if(pattern.getSubject().isVariableField()) {
			KiWiNode binding = match.getBindings().get(pattern.getSubject());
			if(binding != null) {
				result = binding.equals(triple.getSubject());
			} else {
				match.getBindings().put((VariableField) pattern.getSubject(), triple.getSubject());
			}
		}

		if(result && pattern.getProperty().isResourceField()) {
			result = ((ResourceField)pattern.getProperty()).getResource().equals(triple.getProperty());
		} else if(result && pattern.getProperty().isVariableField()) {
			KiWiNode binding = match.getBindings().get(pattern.getProperty());
			if(binding != null) {
				result = binding.equals(triple.getProperty());
			} else {
				match.getBindings().put((VariableField) pattern.getProperty(), triple.getProperty());
			}
		}

		if(result && pattern.getContext() != null && pattern.getContext().isResourceField()) {
			result = ((ResourceField)pattern.getContext()).getResource().equals(triple.getContext());
		} else if(result && pattern.getContext() != null && pattern.getContext().isVariableField()) {
			KiWiNode binding = match.getBindings().get(pattern.getContext());
			if(binding != null) {
				result = binding.equals(triple.getContext());
			} else {
				match.getBindings().put((VariableField) pattern.getContext(), triple.getContext());
			}
		}

		if(result && pattern.getObject().isResourceField()) {
			result = ((ResourceField)pattern.getObject()).getResource().equals(triple.getObject());
		} else if(result && pattern.getObject().isLiteralField()) {
			result = ((LiteralField)pattern.getObject()).getLiteral().equals(triple.getObject());
		} else if(result && pattern.getObject().isVariableField()) {
			KiWiNode binding = match.getBindings().get(pattern.getObject());
			if(binding != null) {
				result = binding.equals(triple.getObject());
			} else {
				match.getBindings().put((VariableField) pattern.getObject(), triple.getObject());
			}
		}


		if(result) {
			match.getJustifications().add(triple);
			return match;
		} else {
			return null;
		}
	}

	@PreDestroy
	public void shutdown() {
        log.info("shutting down reasoning service ...");
		reasoners.shutdown();
		try {
			if(!reasoners.awaitTermination(60,TimeUnit.SECONDS)) {
				log.error("forcibly interrupting reasoning; you may need to re-run the reasoner to ensure data consistency");
				reasoners.shutdownNow();
			}
		} catch (InterruptedException e) {
		}

	}


}
